package com.liu;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

/**
 * 生产者 --
 */
public class Productor {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        //1.创建连接工厂  设置连接信息
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.254.131");//连接的mq服务器地址
        factory.setUsername("root");//默认市guest
        factory.setPassword("123123");//默认市guest
        factory.setPort(5672);//连接端口号
        factory.setVirtualHost("erp");//虚拟主机  默认是 /

        //2.用连接工厂创建连接
        Connection conn = factory.newConnection();

        //3.创建信道
        Channel channel = conn.createChannel();

        //4.开启发布确认
        channel.confirmSelect();

        //声明队列名称
        String queue1 = "confirm-async";

        //5.声明队列
        /**
         * queue  队列名称
         * durable  是否持久化  true 队列持久化 mq服务器重启仍然保留 false队列名就会消失
         * exclusive  是否独占连接 true 信道必须是同一个连接创建的才 能读取队列中的消息
         * autoDelete  是否自动删除队列  （前提是队列不再使用）
         * arguments  向队列传递的其他参数
         */
        channel.queueDeclare(queue1,false,false,false,null);

        //跳表  适用于高并发  绑定发布的序列号与消息
        ConcurrentNavigableMap<Long,String> skipListMap = new ConcurrentSkipListMap<>();
        ConcurrentNavigableMap<Long,String> failListMap = new ConcurrentSkipListMap<>();

        //增加监听  等待服务器确认或返回
        channel.addConfirmListener((deliveryTag,multiple)->{
            System.out.println("确认回调"+multiple+"-------------"+deliveryTag);
            if(multiple){
                //为true  表示所有的消息的序号小于等于deliverTag的消息  得到服务器的确认
                //返回key值小于等于deluverTag的集合 子集,然后进行移除  不需要遍历调表
                ConcurrentNavigableMap childMap= skipListMap.headMap(deliveryTag);
                //将子集从调表中移除
                childMap.clear();
            }else{
                //false 消息表示delivertag的消息  服务器已经确认
                skipListMap.remove(deliveryTag);
            }
        },(deliveryTag,multiple)->{
            System.out.println("否定回调"+multiple+"-------------"+deliveryTag);
            //否定的消息，没有发布成功的消息
            if(multiple){
                ConcurrentNavigableMap childMap = skipListMap.headMap(deliveryTag);
                failListMap.putAll(childMap);
            }else{
                //将失败的消息记录
                failListMap.put(deliveryTag,skipListMap.get(deliveryTag));
            }

        });

        for (int i = 0; i < 1000; i++) {
            String msg = "消息"+i;
            Long seq  = channel.getNextPublishSeqNo();
            //将序列号与消息关联起来
            skipListMap.put(seq,msg);
            //发布消息
            channel.basicPublish("",queue1,MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes());
        }



        //如果跳表中还有消息
        if(skipListMap.size()>0){
            System.in.read();
        }

        //8.释放资源
        channel.close();
        conn.close();

    }



}
